package com.shanks.framework.mq.processor;

import com.shanks.framework.mq.common.constant.MQConstant;
import com.shanks.framework.mq.common.util.JsonUtil;
import com.shanks.framework.mq.domain.MQMessage;
import com.shanks.framework.mq.enums.MessageRecordStatusEnum;
import com.shanks.framework.mq.repository.MessageRecordRepository;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.Map;

/**
 * 消息ack处理器
 */
public class AckMQProcessor implements MQProcessor {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private MessageRecordRepository messageRecordRepository;

    @Override
    public Boolean handleMessage(MQMessage message) throws Exception {
        try {
            String json = message.getBodyAsString();

            Map<String, Object> messageData = JsonUtil.parseToMap(json);
            Boolean result = (Boolean) messageData.get(MQConstant.PROPERTY.RESULT);
            String originKey = (String) messageData.get(MQConstant.PROPERTY.ORIGIN_KEY);
            String clientSource = (String) messageData.get(MQConstant.PROPERTY.SOURCE);

            logger.info("接收到Ack消息, topic:{}, tags:{}, msgId:{}, msgKey:{}, messageUuid:{}, result:{} ", message.getTopic(), message.getTags(),
                    message.getMsgId(), message.getKey(), originKey, result);

            String messageUuid = originKey;

            if (result) {
                messageRecordRepository.updateMessageStatusByUuid(messageUuid, null, MessageRecordStatusEnum.ACKED, new Date(), clientSource);
            } else {
                messageRecordRepository.updateMessageStatusByUuid(messageUuid, null, MessageRecordStatusEnum.CONSUME_FAIL, new Date(), clientSource);
            }
            return true;
        } catch (Exception e) {
            logger.error("处理Ack消息失败", e);
        }
        return false;
    }
}
